package publish

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/streadway/amqp"
	"golang-manage-admin/providers"
)

var PublishObject *Publish

func PublishInit() {
	PublishObject = NewPublish()
}

func NewPublish() *Publish {
	publish := new(Publish)

	publish.init()

	return publish
}

type Publish struct {
	receptionChannel map[string]chan publishChannelMessage
}

// 推送 channel 消息
type publishChannelMessage struct {
	body         []byte
	routingKey   string
	exchangeName string
	headers      amqp.Table
	expiration   string // Message TTL (单位：毫秒)
}

// 队列推送结构体
type PublishMessage struct {
	Api            string      `json:"api"`
	Version        string      `json:"version"`
	TimeStamp      int64       `json:"timestamp"`
	FromPublishKey int         `json:"from_publish_key"`
	DelayedType    string      `json:"delayed_type"` // 延时处理类型 "retry":重试
	Data           interface{} `json:"data"`
}

/**
 * 发送数据,按照指定格式发送数据 -- 直接推送
 */
func (p *Publish) PushFormat(publishKey int, data interface{}) error {
	return p.PushFormatWithHeader(publishKey, data, nil)
}

/**
 *  @Author yang
 *  @Description 推送消息：延时队列 -- 直接推送
 *	@Param publishKey:目标队列（queue 必须是死信：Message TTL 模式： Per-Message TTL）； data:推送内容体； expires:延时推送时间（单位：毫秒）
**/
func (p *Publish) PushFormatDelayed(publishKey int, data interface{}, expires string) error {
	headers := amqp.Table{}
	config := PublishConstantMap[publishKey]
	publishMessage := PublishMessage{
		TimeStamp: time.Now().Unix(),
		Version:   "1",
		Api:       config.api,
		Data:      data,
	}
	// 消息从交换机过期时间,毫秒(x-dead-message插件提供)
	// @Todo 按照最新的文档--header 中设置 x-delay 即过期时间；不用给消息设置过期时间：expiration；
	// 这个版本好像必须设置 expiration （消息过期时间）才会生效
	headers["x-delay"] = expires
	return p.pushFormBase(&publishMessage, publishKey, headers, "0")
}

/**
 * 发送数据
 */
func (p *Publish) PushFormatWithHeader(publishKey int, data interface{}, header amqp.Table) error {
	config := PublishConstantMap[publishKey]
	publishMessage := PublishMessage{
		TimeStamp: time.Now().Unix(),
		Version:   "1",
		Api:       config.api,
		Data:      data,
	}
	return p.pushFormBase(&publishMessage, publishKey, header, "")
}

/**
 *  @Author yang
 *  @Description 推送到死信队列（实现消息延时回推消费）
 * 				作用： 当出现网络波动等不可控制原因消息回推
 *  @Param toPublishKey: 推送目标；fromPublishKey: 推送源； msg : 推送内容对象
**/
func (p *Publish) PushFormRetryMessage(toPublishKey int, fromPublishKey int, msg amqp.Delivery, data interface{}) error {

	// @Todo 推送次数(记录推送次数)
	var retryTimes int32 = 0
	headers := amqp.Table{}
	_ = providers.Logger.Errorf("延时消息处理-消息头部解析:%+v", msg.Headers)
	if _, exist := msg.Headers["retry_times"]; exist {
		retryTimes = msg.Headers["retry_times"].(int32)
		headers = msg.Headers
	}
	headers["retry_times"] = retryTimes

	// @Todo 推送目的地（不同时间的死信队列）
	toConfig := PublishConstantMap[toPublishKey]
	publishMessage := PublishMessage{
		TimeStamp:      time.Now().Unix(),
		Version:        "1",
		Api:            toConfig.api,
		FromPublishKey: fromPublishKey,
		DelayedType:    "retry",
		Data:           data,
	}

	return p.pushFormBase(&publishMessage, toPublishKey, headers, "")
}

/**
 * 推送消息
 * @Param publishMessage 推送消息体；publishKey 目标队列； expiration 目标为延时队列，消息配置延时时间（单位：毫秒）
 */
func (p *Publish) pushFormBase(publishMessage *PublishMessage, publishKey int, header amqp.Table, expiration string) error {
	// 读取配置信息
	config := PublishConstantMap[publishKey]
	pushData, err := json.Marshal(publishMessage)
	if err != nil {
		return err
	}
	// 会有多个 channel 监听分摊推送压力
	p.receptionChannel[config.region] <- publishChannelMessage{
		body:         pushData,
		routingKey:   config.routingKey,
		exchangeName: config.exchangeName,
		headers:      header,
		expiration:   expiration,
	}
	return nil
}

/**
* 创建链接池
 */
func (p *Publish) init() {
	//publishMessageConfig := provviders.Config.Sub("publish.message")
	regionMap := map[string]bool{}

	// @TODO 是否需要声明 exchange 保留
	// 1:如果初始化连接不声明 exchange，推送消息时，不存在exchange（消费者没有初始化exchange），会导致推送消息的时候丢失
	// 2:如果统一生产者生成 exchange 会导致初始化定义的exchange可能会很多，有些业务很少用到的也需要声明（个人觉得这样有点浪费：但更合理--php-fpm模式除外）

	// 提取出所有的 mq 配置，并且新增mq
	for _, value := range PublishConstantMap {
		if providers.Config.IsSet("rabbitmq." + value.region) {
			regionMap[value.region] = true
		}
	}

	p.receptionChannel = map[string]chan publishChannelMessage{}

	for region, _ := range regionMap {
		// 初始化 channel
		p.receptionChannel[region] = make(chan publishChannelMessage)

		// 建立多个连接,这样子往对应的 channel push 数据时，会有多个连接来处理
		for i := 0; i < providers.Config.GetInt("publish.limit"); i++ {
			go p.pushListener(region)
		}
	}
}

/**
 * 推送监听
 */
func (p *Publish) pushListener(region string) {
	regionConn, regionChannel := providers.NewRabbitmq(region)

	// channel close 则关闭连接
	for channelMessage := range p.receptionChannel[region] {
		p.pushMessage(channelMessage, regionChannel)
	}

	// 结束之后要重新新建
	err := regionChannel.Close()
	if err != nil {
		_ = providers.Logger.Errorf("关闭 channel 失败; err:%s", err.Error())
	}
	err = regionConn.Close()
	if err != nil {
		_ = providers.Logger.Errorf("关闭 conn 失败; err:%s", err.Error())
	}

	// 休眠 2 秒之后重新建立连接
	time.Sleep(2 * time.Second)
	p.pushListener(region)
}

/**
 * 推送消息
 */
func (p *Publish) pushMessage(channelMessage publishChannelMessage, regionChannel *amqp.Channel) {
	var err error
	for i := 0; i < 3; i++ {
		publishing := amqp.Publishing{
			ContentType: "application/json",
			Body:        channelMessage.body,
			Headers:     channelMessage.headers,
		}
		// 设置消息过期时间
		if len(channelMessage.expiration) > 0 {
			publishing.Expiration = channelMessage.expiration
		}
		err = regionChannel.Publish(
			channelMessage.exchangeName,
			channelMessage.routingKey,
			true,
			false,
			publishing,
		)

		if err == nil {
			break
		}

		// 重试两次之后休息 1 秒之后再重试
		if i == 1 {
			time.Sleep(1 * time.Second)
		}
	}

	// 3次推送失败之后的处理
	if err != nil {
		loggerError := providers.Logger.Errorf(
			"推送消息失败:%s; exchange:%s; routing_key:%s; data: %s",
			err.Error(),
			channelMessage.exchangeName,
			channelMessage.routingKey,
			string(channelMessage.body),
		)

		if loggerError != nil {
			fmt.Printf(
				"推送消息失败:%s; exchange:%s; routing_key:%s; data: %s; logger_err:%s",
				err.Error(),
				channelMessage.exchangeName,
				channelMessage.routingKey,
				string(channelMessage.body),
				loggerError.Error(),
			)
		}
	}
}
